#include <pthread.h>
#include <errno.h>
#include <stdio.h>
#include <netdb.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <unistd.h>
#include <stdio.h>

#include "psd_conn.h"
#include "psd_protocols.h"
#include "psd_conn_mx.h"
#include "psd_logger.h"
#include "psd_tpool.h"
#include "psd_event.h"
#include "psd_config.h"
#include "psd_modules.h"

#include "myriexpress.h"

static psdConn_mxData *psd_conn_mxdata_alloc(mx_endpoint_t endpoint, mx_endpoint_addr_t ep_addr);
static int psd_conn_mx_write(psdConn *sink, const void *buf, size_t len, int flags);
static int psd_conn_mx_set_session_status(psdConn *sink, int status);
static int psd_conn_mx_read(psdConn *src, void *buf, size_t len, int flags);
static int psd_conn_mx_shutdown(psdConn *conn, uint8_t direction);
static int psd_conn_mx_setbufsize(psdConn *conn, uint8_t direction, int size);
static int psd_conn_mx_settimeout(psdConn *conn, uint8_t direction, int seconds);
static void psd_conn_mx_free(psdConn *conn);


psdConn *psd_conn_mx_alloc(mx_endpoint_t endpoint, mx_endpoint_addr_t ep_addr) {
	psdConn *new_conn;
	char buf[10];

	new_conn = psd_conn_alloc();
	if (!new_conn)
		goto error_exit;

	new_conn->conn_private = psd_conn_mxdata_alloc(endpoint,ep_addr);
	if (!new_conn->conn_private)
		goto error_exit2;

	new_conn->stats_private = psd_conn_stats_alloc_def();
	if (!new_conn->stats_private) {
		psd_err(0, "couldn't allocate tcp stats structure");
		goto error_exit3;
	}

	new_conn->protocol = "MX";

	uint64_t nic_id;
	uint32_t endpoint_id;
	mx_decompose_endpoint_addr(ep_addr,&nic_id,&endpoint_id);
	
	sprintf(buf, "%d", endpoint_id);
	new_conn->description = strdup(buf);
	new_conn->read2 = psd_conn_mx_read;
	new_conn->write2 = psd_conn_mx_write;
	new_conn->shutdown2 = psd_conn_mx_shutdown;
	new_conn->setbufsize2 = psd_conn_mx_setbufsize;
	new_conn->settimeout2 = psd_conn_mx_settimeout;
	new_conn->get_stat2 = psd_conn_default_get_stat;
	new_conn->free_conn_private2 = psd_conn_mx_free;
	new_conn->free_stats2 = psd_conn_free_stats_def;
	new_conn->send_msg2 = psd_conn_default_send_msg;
	new_conn->get_msg2 = psd_conn_default_get_msg;
	new_conn->set_session_status2=psd_conn_mx_set_session_status;
	return new_conn;

error_exit4:
	free(new_conn->stats_private);
error_exit3:
	free(new_conn->conn_private);
error_exit2:
	free(new_conn);
error_exit:
	return NULL;
}

psdConn_mxData *psd_conn_mxdata_alloc(mx_endpoint_t endpoint, mx_endpoint_addr_t ep_addr) {
	psdConn_mxData *new_mxdata;

	new_mxdata = (psdConn_mxData *) malloc(sizeof(psdConn_mxData));
	if (!new_mxdata)
		goto error_exit;

	if (pthread_mutex_init(&(new_mxdata->lock), 0) < 0) {
		goto error_exit_mxdata;
	}

	new_mxdata->hdr_bufs = malloc(sizeof(psdmxHeader) * 1000);
	if (!new_mxdata->hdr_bufs)
		goto error_exit_mutex;

	new_mxdata->endpoint = endpoint;
	new_mxdata->ep_addr = ep_addr;
	new_mxdata->closed = 0;
	new_mxdata->read_closed = 0;
	new_mxdata->write_closed = 0;
	new_mxdata->left_to_write = 0;
	new_mxdata->left_to_read = 0;
	new_mxdata->hdr_bufs_head = 0;
	new_mxdata->session_established=0;
	new_mxdata->close_msg_received=0;
	return new_mxdata;

error_exit_mutex:
	pthread_mutex_destroy(&new_mxdata->lock);
error_exit_mxdata:
	free(new_mxdata);
error_exit:
	return NULL;
}

static int psd_conn_mx_set_session_status(psdConn *sink, int status)
{
  psd_info(10, "bakalim dombili buraya geliyo mu1");

  psdConn_mxData *mx_data = (psdConn_mxData *) sink->conn_private;
  psd_info(10, "bakalim dombili buraya geliyo mu2");
  mx_data->session_established=status;
}

static int psd_conn_mx_write(psdConn *sink, const void *buf, size_t len, int flags) {
	psdConn_mxData *mx_data = (psdConn_mxData *) sink->conn_private;
	int sent;
	mx_segment_t segment;
	mx_request_t req;
	mx_status_t status;
	uint32_t result;
	mx_return_t retval;


	psd_info(10, "mx_write() ------------------------------------------------- 0x%llx",*((uint64_t *)buf));
	
	pthread_mutex_lock(&(mx_data->lock));
	{
	  if(mx_data->session_established == 0)
	    {
		psdmxHeader *hdr = &(mx_data->hdr_bufs[mx_data->hdr_bufs_head]);

		mx_data->hdr_bufs_head++;
		mx_data->hdr_bufs_head %= 950;

		hdr->type = MX_TYPE_DATA;
		hdr->args = len;

		if (mx_data->write_closed) {
			sent = -1;
		} else {
			psd_info(10, "Writing header");
			uint32_t amt_to_send = sizeof(psdmxHeader);
			uint32_t amt_sent = 0;
			do {
				segment.segment_ptr=hdr;
				segment.segment_length=sizeof(psdmxHeader);
				psd_info(10, "----------------------------------------1 %d %d %d",sizeof(psdmxHeader),hdr->type,hdr->args);
				retval = mx_issend(mx_data->endpoint,&segment,1,mx_data->ep_addr,2,NULL,&req);
				if(retval!=MX_SUCCESS)
				  {
				    psd_err(0, "mx_issend could not achive sending packet header");				    
				  }
				psd_info(10, "----------------------------------------2");

				retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
				psd_info(10, "----------------------------------------3");

				if(retval!=MX_SUCCESS)
                                  {
                                    psd_err(0, "MX wait not achive sending packet header");
                                  }
				psd_info(10, "----------------------------------------4");

				psd_info(10, "Sent: %d", result);
				if (result > 0) {
				        amt_sent = status.xfer_length;
					amt_to_send -=amt_sent;
					psd_info(10, "Amt_sent: %d", amt_sent);
				} else {
					psd_info(10, "Failed to write header: %d", result);
					break;
				}
			} while (amt_to_send > 0);

			psd_info(10, "Amt_sent: %d", amt_sent);
			if (amt_sent == sizeof(psdmxHeader)) {
			        psd_info(10, "Wrote header. Writing xsp header: %ld",  sizeof(xspMsgHdr));
				((psdConn_defStats *)sink->stats_private)->bytes_written += sizeof(psdmxHeader);
				segment.segment_ptr=buf;
				if(len <= sizeof(xspMsgHdr))
				  segment.segment_length=len;
				else
				  segment.segment_length=sizeof(xspMsgHdr);
				retval = mx_issend(mx_data->endpoint,&segment,1,mx_data->ep_addr,3,NULL,&req);
                                if(retval!=MX_SUCCESS)
                                  {
                                    psd_err(0, "mx_issend could not achive sending packets");
                                  }
                                retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
                                if(retval!=MX_SUCCESS)
                                  {
                                    psd_err(0, "MX wait not achive sending packets");
                                  }

				if (result > 0)
					((psdConn_defStats *)sink->stats_private)->bytes_written += status.xfer_length;
				if(status.xfer_length== sizeof(xspMsgHdr))
				  psd_info(10, "Wrote xsp header succesfully. Message length is: %ld",  sizeof(xspMsgHdr));
				else if(status.xfer_length > 0)
				  psd_info(10, "Wrote xsp header partially achieved. Message length is: %ld", status.xfer_length );
				mx_data->left_to_write=len-status.xfer_length;
			}
		}

		if (mx_data->write_closed ) {
		  sent = -1;
                } 
		else if(len > sizeof(xspMsgHdr)){
		  uint32_t amt_to_send = mx_data->left_to_write;
		  uint32_t amt_sent = 0;

		  psd_info(10, "Wrote header and xsp header. Writing body: %ld",  len - sizeof(xspMsgHdr));
		  ((psdConn_defStats *)sink->stats_private)->bytes_written += sizeof(psdmxHeader);
		  segment.segment_ptr=buf+sizeof(xspMsgHdr);
		  segment.segment_length=len-sizeof(xspMsgHdr);
		  
		  retval = mx_issend(mx_data->endpoint,&segment,1,mx_data->ep_addr,3,NULL,&req);
		  if(retval!=MX_SUCCESS)
		    {
		      psd_err(0, "mx_issend could not achive sending packets");
		    }
		  retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
		  if(retval!=MX_SUCCESS)
		    {
		      psd_err(0, "MX wait not achive sending packets");
		    }
		  
		  if (result > 0)
		    ((psdConn_defStats *)sink->stats_private)->bytes_written += status.xfer_length;
		  if(status.xfer_length== len - sizeof(xspMsgHdr))
		    psd_info(10, "Wrote body succesfully. Message length is: %ld",  len - sizeof(xspMsgHdr));
		  else if(status.xfer_length > 0)
		    psd_info(10, "Wrote body partially achieved. Message length is: %ld", status.xfer_length );
                }

	    }
	  else if(mx_data->session_established == 1)
	    {
              psdmxHeader *hdr = &(mx_data->hdr_bufs[mx_data->hdr_bufs_head]);

	      mx_data->hdr_bufs_head++;
	      mx_data->hdr_bufs_head %= 950;

	      hdr->type = MX_TYPE_DATA;
	      hdr->args = len;

	      if (mx_data->write_closed) {
		sent = -1;
	      } else {
		psd_info(10, "Writing header");
		uint32_t amt_to_send = sizeof(psdmxHeader);
		uint32_t amt_sent = 0;
		do {
		  segment.segment_ptr=hdr;
		  segment.segment_length=sizeof(psdmxHeader);
		  psd_info(10, "----------------------------------------1 %d %d %d",sizeof(psdmxHeader),hdr->type,hdr->args);
		  retval = mx_issend(mx_data->endpoint,&segment,1,mx_data->ep_addr,2,NULL,&req);
		  if(retval!=MX_SUCCESS)
		    {
		      psd_err(0, "mx_issend could not achive sending packet header");
		    }
		  psd_info(10, "----------------------------------------2");

		  retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
		  psd_info(10, "----------------------------------------3");

		  if(retval!=MX_SUCCESS)
		    {
		      psd_err(0, "MX wait not achive sending packet header");
		    }
		  psd_info(10, "----------------------------------------4");

		  psd_info(10, "Sent: %d", result);
		  if (result > 0) {
		    amt_sent = status.xfer_length;
		    amt_to_send -=amt_sent;
		    psd_info(10, "Amt_sent: %d", amt_sent);
		  } else {
		    psd_info(10, "Failed to write header: %d", result);
		    break;
		  }
		} while (amt_to_send > 0);

		psd_info(10, "Amt_sent: %d", amt_sent);
		if (amt_sent == sizeof(psdmxHeader)) {
		  psd_info(10, "Wrote header. Writing body: %ld",  len);
		  ((psdConn_defStats *)sink->stats_private)->bytes_written += sizeof(psdmxHeader);
		  segment.segment_ptr=buf;
		  segment.segment_length=len;
		  retval = mx_issend(mx_data->endpoint,&segment,1,mx_data->ep_addr,3,NULL,&req);
		  if(retval!=MX_SUCCESS)
		    {
		      psd_err(0, "mx_issend could not achive sending packets");
		    }
		  retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
		  if(retval!=MX_SUCCESS)
		    {
		      psd_err(0, "MX wait not achive sending packets");
		    }

		  if (result > 0)
		    ((psdConn_defStats *)sink->stats_private)->bytes_written += status.xfer_length;
		  if(status.xfer_length== len)
		    psd_info(10, "Wrote body succesfully. Message length is: %ld",  len);
		  else if(status.xfer_length > 0)
		    psd_info(10, "Wrote body partially achieved. Message length is: %ld", status.xfer_length );
		  mx_data->left_to_write=len-status.xfer_length;
		}
	      }
	    }
	}
	pthread_mutex_unlock(&(mx_data->lock));
	
	return status.xfer_length;
}

static int psd_conn_mx_read(psdConn *src, void *buf, size_t len, int flags) {
	psdConn_mxData *mx_data = (psdConn_mxData *) src->conn_private;
	int recvd;
	uint32_t amt_to_read;
	uint32_t amt_read;
	psdmxHeader hdr;
	mx_segment_t segment[2];
	mx_status_t status;
	mx_request_t req;
	mx_return_t retval;
	uint32_t result;

	psd_info(10, "mx_read()--------------------------------------- %d",mx_data->left_to_read);
	
	if (mx_data->read_closed == 1) {
		psd_err(0, "Tried to read from closed MX socket");
		return -1;
	}

	while (mx_data->left_to_read == 0) {
		amt_to_read = sizeof(hdr);
		amt_read = 0;

		do {
			psd_info(10, "mx_irecv(): in: %d", amt_to_read);
			segment[0].segment_ptr=&hdr;
			segment[0].segment_length=sizeof(psdmxHeader)+1;
			psd_info(10, "--------------------------1 %d",segment[0].segment_length);
			retval = mx_irecv(mx_data->endpoint, segment,2,2,0xffffffffffffffff,NULL,&req);
			psd_info(10, "--------------------------1");
			if(retval!=MX_SUCCESS)
			  {
			    psd_err(0, "mx_irecv could not achive receiving packet header %d",retval);
			  }
			retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
			psd_info(10, "--------------------------2");

			if(retval!=MX_SUCCESS)
			  {
			    psd_err(0, "MX wait not achive sending packet header");
			  }
			psd_info(10, "--------------------------3");
			
			psd_info(10, "mx_recv(): out: %d", status.xfer_length);
			if (result > 0) {
			  amt_to_read -= status.xfer_length;
			  amt_read += status.xfer_length;
			}
			else  {
			  psd_err(0, "Failed to receive header: %d", result);
			  return result;
			}
			
		} while (amt_to_read > 0);

		if (hdr.type == MX_TYPE_CLOSE) {
			uint32_t direction = hdr.args;

			psd_info(10, "Got close message");

			if (direction == PSD_RECV_SIDE) {
				pthread_mutex_lock(&(mx_data->lock));
				{
					mx_data->write_closed = 1;
				}
				pthread_mutex_unlock(&(mx_data->lock));
			} else {
				mx_data->read_closed = 1;
				mx_data->close_msg_received=1;
				return 0;
			}
			mx_data->close_msg_received=1;
		} else if (hdr.type == MX_TYPE_DATA) {
			mx_data->left_to_read = hdr.args;
			psd_info(10, "Got data message for %lu bytes", mx_data->left_to_read);

		} else {
			psd_info(10, "got a junk type: %u", hdr.type);
			return -1;
		}
	}
	
	amt_to_read = mx_data->left_to_read;
	if (amt_to_read > len) {
		amt_to_read = len;
	}

	psd_info(10, "mx_irecv(): in: %d", amt_to_read);

	segment[0].segment_ptr=buf;
	segment[0].segment_length=amt_to_read+1;
	retval = mx_irecv(mx_data->endpoint, segment,2,3,0xffffffffffffffff,NULL,&req);
	if(retval!=MX_SUCCESS)
	  {
	    psd_err(0, "mx_irecv could not achive receiving packet %d",retval);
	  }
	retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
	if(retval!=MX_SUCCESS)
	  {
	    psd_err(0, "MX wait not achive sending packet ");
	  }

	psd_info(10, "mx_recv(): out: %d", status.xfer_length);
	if (result > 0) {
		mx_data->left_to_read -= status.xfer_length;
		((psdConn_defStats *)src->stats_private)->bytes_read += status.xfer_length;
	}

	psd_info(10, "mx recv returned %d", status.xfer_length);
	
	return status.xfer_length;
}

static int psd_conn_mx_shutdown(psdConn *conn, uint8_t direction) {

  psd_info(10, "mx_shutdown() direction : %d %d", direction);
  
	psdConn_mxData *mx_data = (psdConn_mxData *) conn->conn_private;

	pthread_mutex_lock(&(mx_data->lock));
	{
	        if(mx_data->close_msg_received==1)
		  {
		    if(mx_data->closed!=1)
		      {
		        psd_info(11,"shutting down, close msg received, so we are the listener.");
			mx_disconnect(mx_data->endpoint, mx_data->ep_addr);
			mx_data->closed=1;
		      }
		    else
		      psd_info(11,"shutting down, close message received and the endpointpoint connection has already been closed.");
		  }
		else
		  {
		        psd_info(11,"shutting down, this is connection initiator so close msg is being sent");
			psdmxHeader *hdr = &(mx_data->hdr_bufs[mx_data->hdr_bufs_head++]);
			mx_data->hdr_bufs_head %= 950;
			
			hdr->type = MX_TYPE_CLOSE;
			hdr->args = htonl((uint32_t) direction);
			
			mx_segment_t segment;
			mx_request_t req;
			mx_return_t retval;
			mx_status_t status;
			uint32_t result;
			segment.segment_ptr=hdr;
			segment.segment_length=sizeof(psdmxHeader);
			retval = mx_issend(mx_data->endpoint,&segment,1,mx_data->ep_addr,2,NULL,&req);
			if(retval!=MX_SUCCESS)
			  {
			    psd_err(0, "mx_issend could not achive sending packet header");
			  }
			retval = mx_wait(mx_data->endpoint,&req,MX_INFINITE,&status,&result);
			if(retval!=MX_SUCCESS)
			  {
			    psd_err(0, "MX wait not achive sending packet header");
			  }
			mx_disconnect(mx_data->endpoint, mx_data->ep_addr);
			mx_close_endpoint(mx_data->endpoint);
		  }
		
	}
	
	pthread_mutex_unlock(&(mx_data->lock));
	
	return 0;
}

static int psd_conn_mx_setbufsize(psdConn *conn, uint8_t direction, int size) {
	psdConn_mxData *mx_data = (psdConn_mxData *) conn->conn_private;
	int new_bufsize;
	int junk;
	psd_info(10, "mx_setbufsize()");

	return 0;

error_exit:
	return -1;
}

static int psd_conn_mx_settimeout(psdConn *conn, uint8_t direction, int seconds) {
	psd_warn(10, "can't set timeout on mx sockets");
	return 0;
}

static void psd_conn_mx_free(psdConn *conn) {
	psdConn_mxData *mx_data = (psdConn_mxData *) conn->conn_private;

	free(mx_data);
}
